import json
import pandas as pd
import chardet
from datetime import datetime
from pathlib import Path
from typing import Dict, Any, Union, List


class MetadataExtractor:
    """
    从各种文件格式中提取元数据的工具类

    使用示例:
    >>> extractor = MetadataExtractor()
    >>> csv_meta = extractor.from_csv("data/sample.csv")
    >>> json_meta = extractor.from_json("data/sample.json")
    """

    @staticmethod
    def from_csv(file_path: Union[str, Path]) -> Dict[str, Any]:
        """
        从CSV文件提取结构化元数据
        :param file_path: CSV文件路径
        :return: 包含元数据的字典
        """
        file_path = Path(file_path)
        try:
            # 检测文件编码
            with open(file_path, 'rb') as f:
                result = chardet.detect(f.read())
            encoding = result['encoding']

            df = pd.read_csv(file_path, encoding=encoding)
        except Exception as e:
            print(f"读取 CSV 文件时出错: {e}")
            return None

        return {
            "file_info": {
                "path": str(file_path.absolute()),
                "size": file_path.stat().st_size,
                "created_at": datetime.fromtimestamp(file_path.stat().st_ctime).isoformat(),
                "modified_at": datetime.fromtimestamp(file_path.stat().st_mtime).isoformat(),
                "format": "csv"
            },
            "structure": {
                "columns": list(df.columns),
                "dtypes": {col: str(dtype) for col, dtype in df.dtypes.items()},
                "row_count": len(df),
                "sample_data": {
                    col: MetadataExtractor._get_sample_values(df[col])
                    for col in df.columns
                }
            },
            "statistics": {
                col: MetadataExtractor._get_column_stats(df[col])
                for col in df.columns
            }
        }

    @staticmethod
    def from_json(file_path: Union[str, Path]) -> Dict[str, Any]:
        """
        从JSON文件提取结构化元数据
        :param file_path: JSON文件路径
        :return: 包含元数据的字典
        """
        file_path = Path(file_path)
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
        except Exception as e:
            print(f"读取 JSON 文件时出错: {e}")
            return None

        def analyze_structure(obj: Any, path: str = "") -> Dict[str, Any]:
            if isinstance(obj, dict):
                return {
                    "type": "object",
                    "properties": {
                        key: analyze_structure(value, f"{path}.{key}" if path else key)
                        for key, value in obj.items()
                    }
                }
            elif isinstance(obj, list):
                element_types = list(set(
                    analyze_structure(item)["type"]
                    for item in obj[:100]  # 限制检查前100个元素以防性能问题
                )) if obj else ["unknown"]

                return {
                    "type": "array",
                    "element_types": element_types,
                    "length": len(obj)
                }
            else:
                return {
                    "type": type(obj).__name__,
                    "sample_value": obj if MetadataExtractor._is_simple_type(obj) else None
                }

        return {
            "file_info": {
                "path": str(file_path.absolute()),
                "size": file_path.stat().st_size,
                "created_at": datetime.fromtimestamp(file_path.stat().st_ctime).isoformat(),
                "format": "json"
            },
            "structure": analyze_structure(data),
            "estimated_records": MetadataExtractor._estimate_json_records(data)
        }

    @staticmethod
    def _get_sample_values(series: pd.Series, n: int = 5) -> List[Any]:
        """获取列的样本值"""
        non_null = series.dropna()
        return non_null.sample(min(n, len(non_null))).tolist()

    @staticmethod
    def _get_column_stats(series: pd.Series) -> Dict[str, Any]:
        """获取列的统计信息"""
        stats = {
            "null_count": int(series.isnull().sum()),
            "unique_count": int(series.nunique())
        }

        if pd.api.types.is_numeric_dtype(series):
            stats.update({
                "min": float(series.min()),
                "max": float(series.max()),
                "mean": float(series.mean()),
                "std": float(series.std())
            })

        return stats

    @staticmethod
    def _is_simple_type(value: Any) -> bool:
        """检查是否为简单类型（可安全展示）"""
        return isinstance(value, (str, int, float, bool)) or value is None

    @staticmethod
    def _estimate_json_records(data: Any) -> int:
        """估算JSON数据中的记录数"""
        if isinstance(data, list):
            return len(data)
        elif isinstance(data, dict):
            return 1 + sum(
                MetadataExtractor._estimate_json_records(value)
                for value in data.values()
            )
        return 0
